package retail.loader;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.LogManager;
import org.springframework.beans.BeansException;

import com.mysql.jdbc.StringUtils;

public class TableLoader {
	static final org.apache.log4j.Logger logger2 = LogManager
			.getLogger(TableLoader.class);

	final String tableName;
	public int pageSize = 50000;
	String schema;
	int currency = Config.currency;
	Boolean finished = false;
	private long rowsCount = 0;
	public boolean isSharding = true;
	Executor executor;
	AtomicInteger pageIndex = new AtomicInteger(0);
	SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	String descTable;
	Map<String, String> fieldTypes;
	KeyInfo keyInfo;
	String maxId = null;
	String fieldString;
	public String getTableName() {
		return this.tableName;
	}

	public synchronized void count(int val) {
		this.rowsCount += val;
	}

	public synchronized void finish(boolean val) {
		this.finished = val;
	}

	public TableLoader(final String schema, final String tableName) {

		this.tableName = descTable = tableName;
		// if(Config.current.isHugeTable(tableName) && Loader.updateDate ==
		// null)
		// descTable = tableName + "_temp";
		// else
		// descTable = tableName;
		this.schema = schema;
		executor = Executors.newFixedThreadPool(currency);
<<<<<<< HEAD
		this.pageSize =5;// Config.current.getPageSize(tableName);
=======
		this.pageSize = Config.current.getPageSize(tableName);
>>>>>>> 1ce2a0af3d001f8d009b9ae2a3a38d3be863296c
		fieldTypes = getFieldsType();
		this.keyInfo = getKeyInfo();
		
		//getMaxID();
	}

	private String getMaxID() {
		if (keyInfo == null)
			return null;
		String script = String.format("Select max(%s) From %s ", keyInfo.name,
				tableName);
		boolean merge = null != Loader.updateDateMin;//增量
		if (!merge) 
			script += " Where update_time <= '"
					+ formatter.format(Loader.updateDateMax) + "'";
		logger2.debug(script + " -- " + merge);
		Connection con = null;//
		try {
			if( !merge)
				con = Loader.getMysqlConnection();
			else
				con = Loader.getOracleConnection();
			
			PreparedStatement ps = con.prepareStatement(script);
			ps.execute();
			ResultSet rs = ps.getResultSet();
			while (rs.next()) {
				this.maxId = rs.getString(1);
				break;
			}
			rs.close();
			ps.close();
			con.close();
		} catch (Exception e) {
			logger2.error("get maxId types error", e);
		}
		return this.maxId;
	}

	private KeyInfo getKeyInfo() {
		String script = "select COLUMN_NAME,DATA_TYPE from information_schema.`COLUMNS` where TABLE_SCHEMA='"
				+ Config.current.getSchema()
				+ "'  AND COLUMN_KEY='PRI' and COLUMN_NAME = 'id' AND TABLE_NAME='"
				+ tableName + "'";
		Connection con = null;//
		KeyInfo key = new KeyInfo();

		try {
			con = Loader.getMysqlConnection();
			PreparedStatement ps = con.prepareStatement(script);
			ps.execute();
			ResultSet rs = ps.getResultSet();
			while (rs.next()) {
				key.name = rs.getString(1);
				key.type = rs.getString(2);
			}
			rs.close();
			ps.close();
			con.close();
		} catch (Exception e) {
			logger2.error("get key types error", e);
		}
		if (key.name != null)
			return key;
		return null;
	}

	private void clearn() throws Exception {
		Connection con = null;//
		con = Loader.getOracleConnection();
		String script = "delete from " + descTable + " WHERE 1=1 ";
		if (!StringUtils.isEmptyOrWhitespaceOnly(Loader.shardingFlag)) {
			if (isShardingTable(con)) {
<<<<<<< HEAD
				if ("U010101_Z".equalsIgnoreCase(Loader.shardingFlag) || "Z".equalsIgnoreCase(Loader.shardingFlag))
					script += String.format(" AND %s in ('%s) ",Config.current.getShardingFeild(),
							Loader.shardingFlag + "','U010101_Q','0_Z','Z','Q'" );
				else 
					script +=String.format(" AND %s in ('%s') ",Config.current.getShardingFeild(),Loader.shardingFlag);
=======
				if (!"U010101_Z".equalsIgnoreCase(Loader.shardingFlag))
					script +=String.format(" AND %s in ('%s') ",Config.current.getShardingFeild(),Loader.shardingFlag);
				else
					script += String.format(" AND %s in ('%s') ",Config.current.getShardingFeild(),
							Loader.shardingFlag + "','U010101_Q','0_Z')" );
>>>>>>> 1ce2a0af3d001f8d009b9ae2a3a38d3be863296c
			}
			if (null != (Loader.updateDateMin))
				script += " AND update_time>='"
						+ formatter.format(Loader.updateDateMin) + "'";

		}
<<<<<<< HEAD
		logger2.debug(script);
		PreparedStatement ps = con.prepareStatement(script);
		ps.execute();
		ps.close();		
=======
		
		PreparedStatement ps = con.prepareStatement(script);
		ps.execute();
		ps.close();
		logger2.debug(script);
>>>>>>> 1ce2a0af3d001f8d009b9ae2a3a38d3be863296c
		logger2.info("##clean " + tableName + " with " + Loader.shardingFlag);
		con.close();

	}

	private Map<String, String> getFieldsType() {
		Map<String, String> map = new HashMap<String, String>();
		Connection con = null;//
		StringBuffer sb = new StringBuffer();
		try {
			con = Loader.getOracleConnection();
			String script = String
					.format("select COLUMN_NAME,DATA_TYPE from dba_tab_columns where table_name='%s' and owner ='%s'",
							tableName, Config.current.getOracleUsr()
									.toUpperCase());
			PreparedStatement ps = con.prepareStatement(script);
			ps.execute();
			ResultSet rs = ps.getResultSet();
			while (rs.next()) {
				if(rs.getString(2).equals("BLOB"))
					continue;
				if( map.size() > 0)
					sb.append(",");
				sb.append("`" + rs.getString(1) + "`");
				map.put(rs.getString(1), rs.getString(2));				
			}
			fieldString = sb.toString();
			rs.close();
			ps.close();
			con.close();
		} catch (Exception e) {
			logger2.error("get field types error", e);
		}

		return map;
	}

	private boolean isShardingTable(Connection con) throws Exception {
<<<<<<< HEAD
		String sharding = Loader.shardingFlag.length() ==1 ?"ZONE_NO":"SHARDING_FLAG";
		try {
			String script = String.format(" SELECT COLUMN_NAME FROM ALL_TAB_COLUMNS WHERE TABLE_NAME ='%s' AND COLUMN_NAME = '%s' and owner = '%s'",
					tableName.toUpperCase(),sharding, Config.current.getOracleUsr().toUpperCase() );
=======
		try {
			String script = String.format(" SELECT COLUMN_NAME FROM ALL_TAB_COLUMNS WHERE TABLE_NAME ='%s' AND COLUMN_NAME = 'SHARDING_FLAG' and owner = '%s'",
					tableName.toUpperCase(),Config.current.getOracleUsr().toUpperCase() );
>>>>>>> 1ce2a0af3d001f8d009b9ae2a3a38d3be863296c
			PreparedStatement ps = con.prepareStatement(script);
			ps.execute();
			boolean exist = false;
			ResultSet rs = ps.getResultSet();
			if (rs.next())
<<<<<<< HEAD
				isSharding = exist = sharding.equalsIgnoreCase(rs
=======
				isSharding = exist = "SHARDING_FLAG".equalsIgnoreCase(rs
>>>>>>> 1ce2a0af3d001f8d009b9ae2a3a38d3be863296c
						.getString(1));
			rs.close();
			ps.close();
			return exist;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		}
	}

	boolean haserror = false;

	public boolean run() {
		if (Config.delete && null == Loader.updateDateMin) {
			try {
				clearn();
			} catch (Exception e) {
				logger2.error("delete fail", e);
			}
		}

		while (!this.finished) {
			final CountDownLatch signal = new CountDownLatch(currency);

			for (int i = 0; i < currency; i++) {
				executor.execute(getWorker(signal, pageIndex.get()));
				pageIndex.getAndIncrement();
			}

			try {
				signal.await();
			} catch (InterruptedException e) {
				this.haserror = true;
				this.finish(true);
			}

			if (finished) {
				logger2.info(tableName + " load completed at page index "
						+ pageIndex.get());
				break;
			}
		}
		if (!haserror)
			Loader.setStatus(tableName, rowsCount);
		return haserror;

	}

	Worker getWorker(CountDownLatch signal, int pageIndex) {
		return new Worker(this, signal, pageIndex);
	}
}
